package com.wolfking.aggregate;

import com.google.common.collect.Maps;
import org.springframework.context.ApplicationContext;

import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * <p>
 *
 * @author 赵伟伟(wolfking)
 * created on 2019-07-10 14:08
 */
public class AgggregateStatic {

    private static final AtomicLong THREAD_COUNT = new AtomicLong(0);

    public static final String THREAD_POOL_NAME = "aggregateThreadPool";

    public static final ExecutorService executorService = new ThreadPoolExecutor(1000, 1000, 10, TimeUnit.SECONDS, new SynchronousQueue<>(),
            (runnable) -> {
                Thread thread = new Thread(runnable);
                thread.setName("agggregate-" + THREAD_COUNT.getAndIncrement());
                return thread;
            }, new ThreadPoolExecutor.CallerRunsPolicy());

    private static final Map<String, ExecutorService> executorServiceMap = Maps.newHashMap();


    private static synchronized ThreadPoolExecutor createThreadPoolExecutor(AggregateThreadPool aggregateThreadPool, ApplicationContext acc) {
        RejectedExecutionHandler rejectedExecutionHandler;
        if ("CallerRunsPolicy".equalsIgnoreCase(aggregateThreadPool.rejectedHandler())) {
            rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();
        } else if ("AbortPolicy".equalsIgnoreCase(aggregateThreadPool.rejectedHandler())) {
            rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();
        } else if ("DiscardPolicy".equalsIgnoreCase(aggregateThreadPool.rejectedHandler())) {
            rejectedExecutionHandler = new ThreadPoolExecutor.DiscardPolicy();
        } else if ("DiscardOldestPolicy".equalsIgnoreCase(aggregateThreadPool.rejectedHandler())) {
            rejectedExecutionHandler = new ThreadPoolExecutor.DiscardOldestPolicy();
        } else {
            rejectedExecutionHandler = acc.getBean(aggregateThreadPool.rejectedHandler(), RejectedExecutionHandler.class);
        }
        AtomicInteger threadCount = new AtomicInteger(0);
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(aggregateThreadPool.coreSize(),
                aggregateThreadPool.maxSize(), aggregateThreadPool.keepAlive(),
                aggregateThreadPool.keepAliveUnit(), new ArrayBlockingQueue<>(aggregateThreadPool.queueSize()),
                (r) -> {
                    Thread thread = new Thread(r);
                    thread.setName(aggregateThreadPool.name() + "-" + threadCount.getAndIncrement());
                    return thread;
                }, rejectedExecutionHandler);
        threadPoolExecutor.allowCoreThreadTimeOut(true);
        return threadPoolExecutor;
    }

    public static synchronized ExecutorService getConfigExecutorService(ApplicationContext acc, AggregateThreadPool aggregateThreadPool) {
        String name = aggregateThreadPool.name();
        ExecutorService executorService;
        if (AgggregateStatic.executorServiceMap.containsKey(name)) {
            executorService = AgggregateStatic.executorServiceMap.get(name);
        } else {
            Map<String, ExecutorService> executorServiceMap = acc.getBeansOfType(ExecutorService.class);
            executorService = executorServiceMap.getOrDefault(name, createThreadPoolExecutor(aggregateThreadPool, acc));
            AgggregateStatic.executorServiceMap.put(name, executorService);
        }
        return executorService;
    }
}
